package china.nanjing.mykingdom.serviceImpl;

import china.nanjing.mykingdom.common.EnvoyMQ;
import china.nanjing.mykingdom.common.PolledEnvoyMQ;
import china.nanjing.mykingdom.service.EnvoyService;
import com.alibaba.fastjson.JSON;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.messaging.Message;

import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import java.io.IOException;

/**
 * @author Chen Hongyu
 */
// https://docs.spring.io/spring-cloud-stream/docs/current/reference/htmlsingle/#spring-cloud-streams-overview-using-polled-consumers
@EnableBinding(value = {EnvoyMQ.class, PolledEnvoyMQ.class})
@Service
@Slf4j
public class EnvoyServiceImpl implements EnvoyService {

    @Autowired
    @Output(EnvoyMQ.ENVOY_OUT)
    private MessageChannel channel;

    @Autowired
    PolledEnvoyMQ polledEnvoyMQ;

    /**
     * 发送消息
     *
     * @param message 消息
     */
    @Override
    public void sendMessage(String message) {
        Message<String> stringMessage = MessageBuilder.withPayload(message).setHeaderIfAbsent("routeHeader", 1024).build();
        channel.send(stringMessage);
    }

    /**
     * 消费者轮询处理
     */
    @Scheduled(fixedDelay = 5_000)
    public void poll() {
        polledEnvoyMQ.destIn().poll(message -> {
            byte[] bytes = (byte[]) message.getPayload();
            String payload = new String(bytes);
            log.info("EnvoyServiceImpl poll: " + payload);
        });
    }


    /**
     * 消费者处理消息
     *
     * @param message 消息
     */
    @Override
    @StreamListener(EnvoyMQ.ENVOY_IN)
    public void handleMessage(Message<String> message) {
        Channel channel = (com.rabbitmq.client.Channel) message.getHeaders().get(AmqpHeaders.CHANNEL);
        Long deliveryTag = (Long) message.getHeaders().get(AmqpHeaders.DELIVERY_TAG);
        try {
            if (channel != null && deliveryTag != null) {
                channel.basicAck(deliveryTag, false);
            }
        } catch (IOException e) {
            log.error("EnvoyServiceImpl Error ", e);
            e.printStackTrace();
        }
        log.info("EnvoyServiceImpl handleMessage: " + JSON.toJSONString(message.getPayload()));
    }
}
